package com.darrenchan.spark

import org.apache.spark.streaming.scheduler._
import org.slf4j.LoggerFactory

/**
  * @Desc
  * @Author chenchi03
  * @CreateTime 2019-11-07 21:24
  */
class BjhListener(private val appName:String, private val duration: Int) extends StreamingListener{

    private val logger = LoggerFactory.getLogger("BjhListener")

    override def onReceiverStarted(receiverStarted: StreamingListenerReceiverStarted): Unit = {
        super.onReceiverStarted(receiverStarted)
    }

    override def onReceiverError(receiverError: StreamingListenerReceiverError): Unit = super.onReceiverError(receiverError)

    override def onReceiverStopped(receiverStopped: StreamingListenerReceiverStopped): Unit = super.onReceiverStopped(receiverStopped)

    override def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted): Unit = {
        super.onBatchSubmitted(batchSubmitted)
        val batchInfo = batchSubmitted.batchInfo
        val batchTime = batchInfo.batchTime
        logger.info("BjhListener  batchTime : ", batchTime)
    }

    override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit = {
        val batchInfo = batchStarted.batchInfo
        val processingStartTime = batchInfo.processingStartTime
        logger.info("BjhListener  processingStartTime : ", processingStartTime)
    }

    override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
        val batchInfo = batchCompleted.batchInfo
        val processingStartTime = batchCompleted.batchInfo.processingStartTime

        val processingEndTime = batchInfo.processingEndTime
        val processingDelay = batchInfo.processingDelay
        val totalDelay = batchInfo.totalDelay

        println(batchInfo.numRecords)

        if(processingDelay.get >= 6 * duration * 1000 && totalDelay.get >= 10 * duration * 1000){
            val monitorTitle = s"spark streaming $appName 程序阻塞异常警告"
            val monitorContent = s"BjhListener : processingStartTime -> ${processingStartTime.get}, processingEndTime -> ${processingEndTime.get} , " +
                    s"processingDelay -> ${processingDelay.get} , totalDelay -> ${totalDelay.get}, 请及时检查！"
//            EmailSender.sendMail(monitorTitle, monitorContent)
        }
        logger.info("BjhListener  processingEndTime : ", processingEndTime)
        logger.info("BjhListener  processingDelay : ", processingDelay)
        logger.info("BjhListener  totalDelay : ", totalDelay)
    }

    override def onOutputOperationStarted(outputOperationStarted: StreamingListenerOutputOperationStarted): Unit =
        super.onOutputOperationStarted(outputOperationStarted)

    override def onOutputOperationCompleted(outputOperationCompleted: StreamingListenerOutputOperationCompleted): Unit =
        super.onOutputOperationCompleted(outputOperationCompleted)

}
